package cn.edu.recommender

import com.mongodb.casbah.Imports.MongoClient
import com.mongodb.casbah.MongoClientURI
import com.mongodb.casbah.commons.MongoDBObject
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable


// 连接助手对象
object ConnHelper extends Serializable {
  lazy val jedis = new Jedis("localhost")
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://localhost:27017/recommender"))
}

case class MongoConfig(uri: String, db: String)

// 标准推荐
case class Recommendation(productId: Int, score: Double)

// 用户的推荐
case class UserRecs(userId: Int, recs: Seq[Recommendation])

//商品的相似度
case class ProductRecs(productId: Int, recs: Seq[Recommendation])


object StreamingRecommender {

  val MAX_USER_RATINGS_NUM = 20
  val MAX_SIM_PRODUCTS_NUM = 20
  val MONGODB_STREAM_RECS_COLLECTION = "StreamRecs"
  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_PRODUCT_RECS_COLLECTION = "ProductRecs"


  def main(args: Array[String]): Unit = {
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27017/recommender",
      "mongo.db" -> "recommender",
      "kafka.topic" -> "recommender"
    )

    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 创建一个 SparkSession
    val spark = SparkSession.builder().appName("StatisticsRecommender").master(config("spark.cores")).getOrCreate()
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(2))

    // 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包进行支持
    import spark.implicits._

    // 广播商品相似度矩阵
    //装换成为 Map[Int, Map[Int,Double]]
    val simProductsMatrix = spark
      .read
      .option("uri", config("mongo.uri"))
      .option("collection", MONGODB_PRODUCT_RECS_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[ProductRecs]
      .rdd
      .map { recs => (recs.productId, recs.recs.map(x => (x.productId, x.score)).toMap)
      }.collectAsMap()


    val simProductsMatrixBroadCast = sc.broadcast(simProductsMatrix)

    //创建到 Kafka 的连接
    val kafkaPara = Map(
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest"
    )

    val kafkaStream = KafkaUtils
      .createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Array(config("kafka.topic")), kafkaPara))


    // 产生评分流  UID|MID|SCORE|TIMESTAMP
    val ratingStream = kafkaStream.map { case msg =>
      var attr = msg.value().split("\\|")
      (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt)
    }

    // 核心实时推荐算法
    ratingStream.foreachRDD(
      rdd => rdd.map {
        case (userId, productId, score, timestamp) =>
          println(">>>>>>>>>>>>>>>>")

          //获取当前最近的 M 次商品评分
          val userRecentlyRatings = getUserRecentlyRating(MAX_USER_RATINGS_NUM, userId, ConnHelper.jedis)

          //获取商品 P 最相似的 K 个商品
          val simProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId, userId, simProductsMatrixBroadCast.value)

          //计算待选商品的推荐优先级
          val streamRecs = computeProductScores(simProductsMatrixBroadCast.value, userRecentlyRatings, simProducts)


          //将数据保存到 MongoDB
          saveRecsToMongoDB(userId, streamRecs)
      }
        .count()
    )


    //启动 Streaming 程序
    ssc.start()
    ssc.awaitTermination()

  }

  import scala.collection.JavaConversions._

  def getUserRecentlyRating(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {
    //从用户的队列中取出 num 个评分
    jedis.lrange("userId:" + userId.toString, 0, num)
      .map {
        item =>
          val attr = item.split("\\:")
          (attr(0).trim.toInt, attr(1).trim.toDouble)
      }
      .toArray
  }

  def getTopSimProducts(num: Int, productId: Int, userId: Int, simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])(implicit mongConfig: MongoConfig): Array[Int] = {
    //从广播变量的商品相似度矩阵中获取当前商品所有的相似商品
    val allSimProducts = simProducts(productId).toArray
    //获取用户已经观看过得商品
    val ratingExist = ConnHelper
      .mongoClient(mongConfig.db)(MONGODB_RATING_COLLECTION)
      .find(MongoDBObject("userId" -> userId))
      .toArray
      .map(item => item.get("productId").toString.toInt)

    //过滤掉已经评分过得商品，并排序输出
    allSimProducts
      .filter(x => !ratingExist.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(num)
      .map(x => x._1)
  }

  def computeProductScores(simProducts: scala.collection.Map[Int, Map[Int, Double]], userRecentlyRatings: Array[(Int, Double)], topSimProducts: Array[Int]): Array[(Int, Double)] = {
    // 用于保存每一个待选商品和最近评分的每一个商品的权重得分
    val score = mutable.ArrayBuffer[(Int, Double)]()
    // 用于保存每一个商品的增强因子数
    val increMap = mutable.HashMap[Int, Int]()
    // 用于保存每一个商品的减弱因子数
    val decreMap = mutable.HashMap[Int, Int]()

    for (topSimProduct <- topSimProducts; userRecentlyRating <- userRecentlyRatings) {
      val simScore = getProductsSimScore(simProducts, userRecentlyRating._1, topSimProduct)
      if (simScore > 0.6) {
        score += (topSimProduct, simScore * userRecentlyRating._2)
        if (userRecentlyRating._2 > 3) {
          increMap(topSimProduct) = increMap.getOrDefault(topSimProduct, 0) + 1
        } else {
          decreMap(topSimProduct) = decreMap.getOrDefault(topSimProduct, 0) + 1
        }
      }
    }

    score
      .groupBy(_._1)
      .map {
        case (productId, sims) =>
          (productId, sims.map(_._2).sum / sims.length + log(increMap.getOrDefault(productId, 1)) - log(decreMap.getOrDefault(productId, 1)))
      }
      .toArray
      .sortWith(_._2 > _._2)
  }

  def getProductsSimScore(simProducts: scala.collection.Map[Int, Map[Int, Double]], userRatingProduct: Int, topSimProduct: Int): Double = {
    simProducts.get(topSimProduct) match {
      case Some(sim) => sim.get(userRatingProduct) match {
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }

  // 取 10 的对数
  def log(m: Int): Double = {
    math.log(m) / math.log(10)
  }


  def saveRecsToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit = {
    //到 StreamRecs 的连接
    val streaRecsCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_STREAM_RECS_COLLECTION)
    streaRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
    streaRecsCollection.insert(MongoDBObject("userId" -> userId, "recs" -> streamRecs.map(x => MongoDBObject("productId" -> x._1, "score" -> x._2))))
  }

}
